package com.imooc.spark.streaming.windowOperations

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * 上述更有效的版本，reduceByKeyAndWindow()其中使用前一窗口的reduce值逐步计算每个窗口的reduce值。
  * 这是通过减少进入滑动窗口的新数据，并“反向减少”离开窗口的旧数据来完成的。
  * 一个例子是当窗口滑动时“添加”和“减去”键的计数。但是，它仅适用于“可逆减少函数”，
  * 即那些具有相应“反向减少”函数的减函数（作为参数invFunc）。
  * 同样reduceByKeyAndWindow，reduce任务的数量可通过可选参数进行配置。
  * 请注意，必须启用检查点才能使用此操作。
  *
  * @description reduceByKeyAndWindow（func，invFunc，windowLength， slideInterval，[ numTasks ]）
  * @author yuyon26@126.com
  * @date 2018/10/24 20:14
  */
object ReduceByKeyAndWindow2 {
  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    ssc.checkpoint("D:\\spark\\checkpoint")
    ssc.sparkContext.setLogLevel("ERROR")

    val lines = ssc.socketTextStream("192.168.31.30", 9999)

    val fmapLines = lines.flatMap(_.split(" "))
    val tuple = fmapLines.map(word => (word, 1))
    val reduceByKeyAndWindow = tuple.reduceByKeyAndWindow((a: Int, b: Int) => (a + b), (a: Int, b: Int) => (a - b), Seconds(10), Seconds(5))

    reduceByKeyAndWindow.print()

    ssc.start()
    ssc.awaitTermination()
  }
}
